package com.ycky.batchProcess.publish;


import cn.hutool.core.collection.CollUtil;
import com.ycky.batchProcess.*;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @description: 发布服务抽象类
 * @author: cjj
 * @time: 2025/3/18 10:59
 */
@Slf4j
public abstract class AbstractPublishProcessService<T, R> implements BatchProcess<T, R> {

    // 执行中的发布任务集合
    protected final Map<String, Boolean> publishingTasks = new ConcurrentHashMap<>();

    // 发布线程
    protected static ExecutorService executorService = Executors.newFixedThreadPool(10);

    // 发布速率限制器
    private static final GlobalRateLimiter rateLimiter = new GlobalRateLimiter();

    // 发布策略
    public abstract BatchProcessStrategy<T, R> getStrategy();

    // 发布上下文
    protected abstract BatchProcessContext createContext(int total);

    // 分页大小 todo 改为可配置的
    private static final int PAGE_SIZE = 500;

    @Override
    public void batchPublish(BatchProcessScene scene, BatchProcessParam<T> param) {
        String publishId = param.getPublishId();
        log.info("开始发布任务,任务标识：{}， 任务场景: {}", publishId, scene.getType());
        // 检查发布任务是否正在执行
        if (publishingTasks.containsKey(publishId)) {
            log.warn("发布任务正在执行中，不允许重复执行, 任务标识: {}", publishId);
            return;
        }
        // 添加到执行中发布任务集合
        publishingTasks.put(publishId, true);
        // 批量发布流程
        if (scene == BatchProcessScene.BATCH_COUNT_PUBLISH) {
            // 异步执行发布操作
            CompletableFuture.runAsync(() -> executePublish(param), executorService);
        }
        // todo 普通发布流程 不需要进度监测
    }


    /**
     * @description: 执行发布操作
     * @author: cjj
     * @date: 2025/3/18 13:59
     * @param: [param:发布参数]
     **/
    private void executePublish(BatchProcessParam<T> param) {
        // 发布数据
        String code = param.getCode();
        // 发布任务Id
        String publishId = param.getPublishId();
        // 用户Id
        T userId = param.getUserId();
        try {
            // 发布策略: 根据业务场景选择对应的策略(策略由集成方实现)
            BatchProcessStrategy<T, R> strategy = getStrategy();
            // 更新发布状态
            strategy.updatePublishStatus(code);
            // 获取待发布总数
            int publishTotal = strategy.getRecordCount(code);
            // 获取未发布数
            int unPublishedCount = strategy.getUnpublishedCount(code);
            if (unPublishedCount == 0) {
                log.info("任务 {} 无待发布的数据， publishTotal: {}, unPublishedCount: {}", publishId, publishTotal, unPublishedCount);
                // 更新状态为已发布
                strategy.updatePublishStatus(code);
                return;
            }
            int current = publishTotal - unPublishedCount;
            log.info("开始发布记录, publishId: {}, publishTotal: {}, unPublishedCount: {}", publishId, publishTotal, unPublishedCount);

            BatchProcessContext context = createContext(publishTotal);

            // 使用游标分页处理
            AtomicInteger processed = new AtomicInteger(current);
            String cursor = null;

            while (true) {
                // 使用游标获取下一批数据
                BatchProcessStrategy.PageResult<R> pageResult = strategy.getNextBatch(code, cursor, PAGE_SIZE);
                if (pageResult == null || CollUtil.isEmpty(pageResult.getRecords())) {
                    break;
                }
                // 处理当前批次的数据
                publishRecords(publishId, code, pageResult.getRecords(), strategy, processed, context, userId);

                // 更新游标
                cursor = pageResult.getNextCursor();
                if (cursor == null) {
                    log.info("已处理完所有记录, publishId: {}", publishId);
                    break;
                }
                // 检查任务是否已停止
                if (!publishingTasks.containsKey(publishId)) {
                    log.info("任务已停止, publishId: {}", publishId);
                    return;
                }
            }

            // 更新发布状态
            strategy.updatePublishStatus(code);
            log.info("发布完成, publishId: {}, 实际处理记录: {}", publishId, processed.get());

        } catch (Exception e) {
            log.error("发布失败, publishId: {}, error: {}", publishId, e.getMessage(), e);
        } finally {
            // 从执行中发布任务集合中移除
            publishingTasks.remove(publishId);
        }
    }


    private void publishRecords(String publishId, String code, List<R> records, BatchProcessStrategy<T, R> strategy,
                                AtomicInteger processed, BatchProcessContext context, T userId) throws InterruptedException {
        log.info("开始批量发布记录, 数据标识: {}, 期望执行速率: {}条/秒", code, strategy.getPublishRate());

        List<R> successRecords = new ArrayList<>();

        for (R record : records) {
            // 检查任务是否已停止
            if (!publishingTasks.containsKey(publishId)) {
                return;
            }

            long startTime = System.currentTimeMillis();
            // 使用全局速率限制器
            rateLimiter.acquire(strategy.getPublishRate());
            processed.incrementAndGet();
            if (strategy.publishRecord(record, context)) {
                successRecords.add(record);
            }

            // 计算当前批次的实际处理速率
            long currentTime = System.currentTimeMillis();
            double elapsedSeconds = (currentTime - startTime) / 1000.0;
            if (elapsedSeconds > 0) {
                double currentSpeed = 1.0 / elapsedSeconds;
                // 更新上下文中的处理速度
                context.setCurrentSpeed(currentSpeed);
            }
            // 推送进度(由子类实现具体得推送数据)
            strategy.sendProgress(userId, context, processed.get());
        }

        // 批量更新状态
        if (!successRecords.isEmpty()) {
            strategy.batchUpdateStatus(successRecords);
        }
    }
}